package cn.doitedu.rtdw.etl;

import cn.doitedu.rtdw.beans.PageAccessBean;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/12/16
 * @Desc: 学大数据，上多易教育
 * 流量olap支撑： 访问时长分析主题的基础表开发
 **/
public class Job03_traffic_olap_accesstime_base {
    public static void main(String[] args) throws Exception {

        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 构建kafka source，来读取kafka dwd层的行为明细
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                .setGroupId("doit43-04")
                .setTopics("dwd-events")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        DataStreamSource<String> streamSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s");


        // 逻辑加工
        SingleOutputStreamOperator<PageAccessBean> beanStream =
                streamSource.map(dwdJson -> {
                    PageAccessBean bean = JSON.parseObject(dwdJson, PageAccessBean.class);
                    bean.setPage_url(bean.getProperties().getOrDefault("url", null));
                    return bean;
                });

        SingleOutputStreamOperator<PageAccessBean> processed = beanStream.keyBy(PageAccessBean::getSession_id)
                .process(new KeyedProcessFunction<String, PageAccessBean, PageAccessBean>() {

                    ValueState<PageAccessBean> state;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        ValueStateDescriptor<PageAccessBean> descriptor = new ValueStateDescriptor<>("pre", PageAccessBean.class);
                        StateTtlConfig ttlConfig =
                                StateTtlConfig.newBuilder(Time.hours(1))
                                        .updateTtlOnReadAndWrite()  // 状态被访问，则ttl计时重置
                                        .build();
                        descriptor.enableTimeToLive(ttlConfig);  // 开启ttl

                        state = getRuntimeContext().getState(descriptor);  // 根据描述符，获取状态对象
                    }

                    @Override
                    public void processElement(PageAccessBean bean, KeyedProcessFunction<String, PageAccessBean, PageAccessBean>.Context ctx, Collector<PageAccessBean> out) throws Exception {

                        if (state.value() == null) {
                            state.update(bean);
                        }

                        if (bean.getEvent_id().equals("page_load")) {

                            // 用本条数据的action_time，替换 状态中的数据的 action_time,然后输出状态中的数据
                            state.value().setAction_time(bean.getAction_time());
                            out.collect(state.value());   // 这里就是插入了一条虚拟事件

                            // 更新 页面起始时间
                            bean.setPage_start_time(bean.getAction_time());
                            // 再将自己放入状态，表示一次新的页面访问开始了
                            state.update(bean);

                            // 再将自己输出
                            out.collect(bean);
                        } else if (bean.getEvent_id().equals("wake_up")) {

                            // 设置 页面起始时间，并覆盖状态
                            bean.setPage_start_time(bean.getAction_time());
                            state.update(bean);

                            out.collect(bean);

                        } else {

                            // 普通事件，只需要将状态中的action_time更新成自己的，然后输出状态中的数据即可
                            state.value().setAction_time(bean.getAction_time());
                            out.collect(state.value());
                        }

                        // 如果是app_close事件，表明本次session结束，应该清理状态垃圾
                        if (bean.getEvent_id().equals("app_close")) {
                            state.clear();
                        }


                    }
                }).filter(bean -> bean.getPage_url() != null && bean.getPage_start_time() != null);


        // 将加工结果转成 表
        tenv.createTemporaryView("tmp",processed,
                Schema.newBuilder()
                        .column("release_channel", DataTypes.STRING())
                        .column("device_type",DataTypes.STRING())
                        .column("page_type",DataTypes.STRING())
                        .column("page_service",DataTypes.STRING())
                        .column("member_level_id",DataTypes.BIGINT())
                        .column("gender",DataTypes.INT())
                        .column("province",DataTypes.STRING())
                        .column("city",DataTypes.STRING())
                        .column("region",DataTypes.STRING())
                        .column("user_id",DataTypes.BIGINT())
                        .column("session_id",DataTypes.STRING())
                        .column("event_id",DataTypes.STRING())
                        .column("action_time",DataTypes.BIGINT())
                        .column("page_url",DataTypes.STRING())
                        .column("page_start_time",DataTypes.BIGINT())
                        .columnByExpression("rt","to_timestamp_ltz(action_time,3)")
                        .watermark("rt","rt - interval '0' second ")
                        .build());



        // 创建doris访问时长olap物理表的映射表
        tenv.executeSql(
                        " create table traffic_olap_access_timelong_doris(                "+
                        "     dt                   date                                   "+
                        "     ,release_channel     varchar                                "+
                        "     ,device_type         varchar                                "+
                        "     ,page_type           varchar                                "+
                        "     ,page_service        varchar                                "+
                        "     ,member_level_id     bigint                                 "+
                        "     ,gender              int                                    "+
                        "     ,province            varchar                                "+
                        "     ,city                varchar                                "+
                        "     ,region              varchar                                "+
                        "     ,user_id             bigint                                 "+
                        "     ,session_id          varchar                                "+
                        "     ,page_url            varchar                                "+
                        "     ,page_start_time     bigint                                 "+
                        "     ,page_end_time       bigint                                 "+
                        " )  with (                                                       "+
                        "     'connector' = 'doris',                                      "+
                        "     'fenodes' = 'doitedu:8030',                                 "+
                        "     'table.identifier' = 'dws.traffic_olap_access_timelong',    "+
                        "     'username' = 'root',                                        "+
                        "     'password' = 'root',                                        "+
                        "     'sink.label-prefix' = 'doit42_no4'                          "+
                        " )                                                               "
        );


        // 用sql进行统计（groupBy max-min)
        tenv.executeSql(
                        " INSERT INTO traffic_olap_access_timelong_doris "+
                        " SELECT                                                      "+
                        "   to_date(date_format(TO_TIMESTAMP_LTZ(page_start_time,3),'yyyy-MM-dd')) AS dt, "+
                        "   release_channel,                                          "+
                        " 	device_type,                                              "+
                        " 	page_type,                                                "+
                        " 	page_service,                                             "+
                        " 	member_level_id,                                          "+
                        " 	gender,                                                   "+
                        " 	province,                                                 "+
                        " 	city,                                                     "+
                        " 	region,                                                   "+
                        " 	user_id,                                                  "+
                        " 	session_id,                                               "+
                        " 	page_url,                                                 "+
                        " 	page_start_time,                                          "+
                        " 	max(action_time) as page_end_time                         "+
                        " from TABLE(                                                 "+
                        "     TUMBLE(table tmp,descriptor(rt),INTERVAL '1' MINUTE)    "+
                        " )                                                           "+
                        " GROUP BY                                                    "+
                        "   window_start,                                             "+
                        " 	window_end,                                               "+
                        "   release_channel,                                          "+
                        " 	device_type,                                              "+
                        " 	page_type,                                                "+
                        " 	page_service,                                             "+
                        " 	member_level_id,                                          "+
                        " 	gender,                                                   "+
                        " 	province,                                                 "+
                        " 	city,                                                     "+
                        " 	region,                                                   "+
                        " 	user_id,                                                  "+
                        " 	session_id,                                               "+
                        " 	page_url,                                                 "+
                        " 	page_start_time                                           "
        );




        env.execute();

    }
}
